package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

// 从socket读取数据，然后进行单词计数
public class Example1 {
    public static void main(String[] args) {
        // 获取流执行环境上下文
        var env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 指定socket为数据源
        // 先启动`nc -lk 9999`
        // `.setParallelism(1)`表示数据源的并行子任务的数量是1个
        var source = env
                .socketTextStream("localhost", 9999)
                .setParallelism(1);

        // map阶段
        // "hello world" => ("hello", 1); ("world", 1)
        // 一对多的转换我们使用`flatMap`
        var mappedStream = source
                .flatMap(new Tokenizer())
                .setParallelism(1);

        // shuffle阶段
        // 将key相同的数据分到一个小组
        // 将二元组的f0字段设置为key
        var keyedStream = mappedStream.keyBy(r -> r.f0);

        // reduce阶段
        // 将相同key的数据做聚合
        var reducedStream = keyedStream
                .reduce(new WordCount())
                .setParallelism(1);

        // 输出结果
        reducedStream
                .print()
                .setParallelism(1);

        // 提交并执行任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // FlatMapFunction<输入的泛型，输出的泛型>
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        // 集合会收集将要发送的数据
        // 由flink自动将数据向下游发送
        @Override
        public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
            var words = in.split(" ");
            for (var word : words) {
                // 将要发送的二元组收集到集合中
                out.collect(Tuple2.of(word, 1));
            }
        }
    }

    // ReduceFunction为什么只需要一个泛型？
    // 因为输入、累加器和输出的类型一致
    // reduce的输出是新累加器
    public static class WordCount implements ReduceFunction<Tuple2<String, Integer>> {
        // 返回值：新累加器
        // 第一个参数：旧累加器
        // 第二个参数：输入数据
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> accumulator, Tuple2<String, Integer> in) throws Exception {
            // 输入数据和累加器的累加规则
            return Tuple2.of(accumulator.f0, accumulator.f1 + in.f1);
        }
    }
}
